package com.seventeen.mq.proxy.server.proxy.ext;

import com.alibaba.fastjson.JSON;
import com.seventeen.mq.proxy.common.exception.base.BaseException;
import com.seventeen.mq.proxy.common.utils.StringUtils;
import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.ProducerBuilder;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;

/**
 * 5.x 生产者代理实体类
 *
 * @author seventeen
 */
public class NewExtMqProducer extends BaseExtMqProducer {
    private static final Logger LOGGER = LoggerFactory.getLogger(NewExtMqProducer.class);

    private Producer producer;

    @Override
    public void destroy() {
        if (producer != null) {
            LOGGER.info("{}  shutdown", this);
            try {
                producer.close();
            } catch (Exception e) {
                LOGGER.error("{}  shutdown error,", this, e);
            }
        }
    }

    @Override
    public void init() {
        try {
            ClientServiceProvider provider = ClientServiceProvider.loadService();
            ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
            if (StringUtils.isNotBlank(accessKey) && StringUtils.isNotBlank(secretKey)) {
                SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey);
                builder.setCredentialProvider(sessionCredentialsProvider);
            }
            ClientConfiguration configuration = builder.build();
            // 初始化Producer时需要设置通信配置以及预绑定的Topic。
            ProducerBuilder producerBuilder = provider.newProducerBuilder();
            for (String topic : getWithTopics()) {
                producerBuilder.setTopics(topic);
            }
            producer = producerBuilder.setClientConfiguration(configuration)
                    .build();
        } catch (Exception e) {
            LOGGER.error("NewExtMqProducer start fail:{},{}", getProducerProxyGroup(), getEndpoints(), e);
        }
    }

    @Override
    public void sendMsg(ExtSendMessage sendMessage) {
        String topic = sendMessage.getTopic();
        String tag = sendMessage.getTag();
        String keys = sendMessage.getKeys();
        String message = sendMessage.getMessage();
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        try {
            MessageBuilder messageBuilder = provider.newMessageBuilder()
                    .setTopic(topic)
                    // 设置消息索引键，可根据关键字精确查找某条消息。
                    .setKeys(keys)
                    // 设置消息Tag，用于消费端根据指定Tag过滤消息。
                    .setTag(tag)
                    // 消息体。
                    .setBody(message.getBytes(RemotingHelper.DEFAULT_CHARSET));
            //延时消息
            if (Objects.nonNull(sendMessage.getDeliveryTimestamp())) {
                messageBuilder.setDeliveryTimestamp(sendMessage.getDeliveryTimestamp());
            }
            // 普通消息发送。
            Message msg = messageBuilder.build();
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(msg);
            LOGGER.info("send message end, msg {},  sendResult {}", msg, JSON.toJSONString(sendReceipt));
        } catch (Exception e) {
            LOGGER.error("send message error, msg {} ,", JSON.toJSONString(sendMessage), e);
            throw new BaseException("消息发送失败");
        }
    }
}
